package com.bestv.search.engine.compass.extend;

import com.bestv.search.common.cache.SystemCache;
import com.bestv.search.common.dataSource.DataSourceKey;
import com.bestv.search.common.dataSource.DataSourceKeyUtil;
import com.bestv.search.common.model.BizType;
import com.bestv.search.common.model.Info;
import com.bestv.search.common.model.InfoHeader;
import com.bestv.search.common.model.KeySubject;
import com.bestv.search.common.service.BizTypeManager;
import com.bestv.search.common.util.DateUtil;
import com.bestv.search.common.util.StringConst;
import com.bestv.search.engine.command.*;
import com.bestv.search.engine.compass.model.HdInfoLucene;
import com.bestv.search.engine.compass.model.HdPersonLucene;
import org.apache.commons.configuration.ConfigurationException;
import org.apache.commons.configuration.PropertiesConfiguration;
import org.apache.commons.configuration.reloading.FileChangedReloadingStrategy;
import org.apache.lucene.search.FieldCache;
import org.compass.core.*;
import org.compass.gps.ActiveMirrorGpsDevice;
import org.compass.gps.CompassGpsException;
import org.compass.gps.device.hibernate.HibernateGpsDevice;

import java.io.File;
import java.io.IOException;
import java.util.*;

/**
 * @author Michael Wang
 */

// $Id: HibernateGpsDeviceExtend.java 295 2013-08-14 04:54:41Z luo.xiaojun $
public class HibernateGpsDeviceExtend extends HibernateGpsDevice implements ActiveMirrorGpsDevice {
    private final String LAST_INDEX_TIME_FORMAT = "yyyy-MM-dd HH:mm:ss";

    private boolean canStartIncrementIndexThread = true;
    private Compass compass;
    private CompassTemplate compassTemplate;
    private String sortType;
    private String lastIndexTimeFile;
    private InfoCommand infoCommand;
    private KeySubjectCommand keySubjectCommand;
    private PersonCommand personCommand;
    private InfoHeaderCommand infoHeaderCommand;
    private BizTypeManager bizTypeManager;
    private SystemCache systemCache;

    private boolean shouldReopenInfo = false;
    private boolean shouldReopenKey = false;
    private boolean shouldReopenHdInfo = false;

    @Override
    protected void doStart() throws CompassGpsException {
        super.doStart();

        // Set compassTemplate
        if (compass == null) {
            throw new IllegalArgumentException("Must set compass property");
        }
        this.compassTemplate = new CompassTemplateExtend(compass);

        File file = new File(lastIndexTimeFile);
        if (!file.exists()) {
            try {
                file.createNewFile();
            } catch (IOException e) {
                log.error("Create lastIndexTimeFile failed" + lastIndexTimeFile);
                e.printStackTrace();
            }
        }

        // init last time of HD
        initHD();

        DataSourceKey key = DataSourceKeyUtil.convert(systemCache.getLocalDataSourceKey());
        List<BizType> bizTypes = bizTypeManager.getAllBizTypes(key);
        LuceneSearchCommandRefactor.setBizTypeValue(bizTypes);

        log.debug("HibernateGpsDeviceExtend has started!");
    }

    private void initHD() {
        // Update MaxHitCountByTags map for p_type=纪实
        LuceneSearchCommandRefactor.setMaxHitCountByTagsMap(infoCommand
                .getMaxHitCountByTags(StringConst.DOCUMENTARY_CHINESE));
    }

    @Override
    protected void doStop() throws CompassGpsException {
        super.doStop();
        log.debug("HibernateGpsDeviceExtend is stopped!");
    }

    public void performMirroring() throws CompassGpsException {
        if (!canStartIncrementIndexThread) {
            log.info("The index thread is closed or the re-index thread is running , waiting until the thread finished......");
            return;
        }
        if (canStartIncrementIndexThread) {
            setCanStartIncrementIndexThread(!canStartIncrementIndexThread);// (1)
            if (canStartIncrementIndexThread) {
                /**
                 * 说明在执行(1)的时候，恰好被其他线程修改了coreDevice.
                 * canStartIncrementIndexThread的状态为false，
                 * 由此可以断定有一个线程正在执行index或increasingIndex，此时该线程应该等待或者直接返回
                 */
                log.info("The thread is busy, please try later! ");
                return;
            }

            try {

                Date begin = new Date();

                // Do index
                log.info(" Index starting......");
                doIndex();

                // Do index for HD
                log.info(" HD Index starting......");
                doIndexForHD();

                // Updates bizTypes
                DataSourceKey key = DataSourceKeyUtil.convert(systemCache.getLocalDataSourceKey());
                List<BizType> bizTypes = bizTypeManager.getAllBizTypes(key);
                LuceneSearchCommandRefactor.setBizTypeValue(bizTypes);

                Date end = new Date();
                log.info(" Index success,total time: " + (end.getTime() - begin.getTime()) / 1000 + "s");
            } catch (Exception e) {
                log.error("Update data to index failed @ " + new Date());
                e.printStackTrace();
            } finally {
                canStartIncrementIndexThread = true;
            }
        }
    }

    public void doIndex() {
        try {
            shouldReopenInfo = false;
            shouldReopenKey = false;

            // Index for infos
            doIndexForInfos(false, null);
            // Index for keysubjects
            doIndexForKeySubject();
            // Index fro info headers
            doIndexForHeaders(false);
        } catch (Exception e) {
            log.error(" Write data to index failed, due to ");
            e.printStackTrace();
        } finally {
            // should reopen readers
            try {
                if (shouldReopenInfo) {
                    LuceneSearchCommandRefactor.setIndexSearcher(LuceneSearchCommandRefactor.refreshIndexCache("info"));
                    // Updates types cache
                    try {
                        String[] typesInIndex = FieldCache.DEFAULT.getStrings(LuceneSearchCommandRefactor
                                .getIndexSearcher().getIndexReader(), StringConst.TYPE);
                        LuceneSearchCommandRefactor.setTypesInIndex(typesInIndex);
                    } catch (IOException e) {
                        log.error("Get types failed from index, due to exception ");
                        e.printStackTrace();
                    }

                    // Updates bizTypes cache
                    try {
                        String[] bizTypesInIndex = FieldCache.DEFAULT.getStrings(LuceneSearchCommandRefactor
                                .getIndexSearcher().getIndexReader(), StringConst.P_TYPE);
                        LuceneSearchCommandRefactor.setBizTypesInIndex(bizTypesInIndex);
                    } catch (IOException e) {
                        log.error("Get bizTypes failed, due to exception ");
                        e.printStackTrace();
                    }
                }

                if (shouldReopenKey) {
                    LuceneSearchCommandRefactor.setKeySearcher(LuceneSearchCommandRefactor
                            .refreshIndexCache("keysubject"));
                }

            } catch (Exception e) {
                log.error("refresh index cache failed @ " + new Date());
                e.printStackTrace();
            }
        }
    }

    public void doIndexForHD() {
        try {
            // Save personIds that have info on line or off line
            Set<String> personIds = new HashSet<String>();
            shouldReopenHdInfo = false;

            // Index for infos
            doIndexForInfos(true, personIds);

            // Indexing for Person
            log.info("Person indexing starting..... @ " + new Date());
            // Index person that have info on line or off line
            int personLimit = 2000;
            if (personIds.size() > 0) {
                List<HdPersonLucene> persons = personCommand.getLucenePersonByIds(personIds);

                // Start index
                List<HdPersonLucene> indexPersons = new ArrayList<HdPersonLucene>();
                for (HdPersonLucene p : persons) {
                    indexPersons.add(p);
                    if (indexPersons.size() == personLimit) {
                        doIndex(persons);
                        indexPersons = new ArrayList<HdPersonLucene>();
                    }
                }
                if (indexPersons.size() > 0) {
                    doIndex(indexPersons);
                }
            }

            // Index for person
            doIndexForPerson();

            // Index fro headers
            doIndexForHeaders(true);
        } catch (Exception e) {
            log.error(" Write data to index failed, due to ");
            e.printStackTrace();
        } finally {
            // should reopen readers
            try {
                if (shouldReopenHdInfo) {
                    LuceneSearchCommandRefactor.setHdIndexSearcher(LuceneSearchCommandRefactor
                            .refreshIndexCache("hdinfo"));

                    // Updates types cache
                    try {
                        String[] typesInHd = FieldCache.DEFAULT.getStrings(LuceneSearchCommandRefactor
                                .getHdIndexSearcher().getIndexReader(), StringConst.TYPE);
                        LuceneSearchCommandRefactor.setTypesInHd(typesInHd);
                    } catch (IOException e) {
                        log.error("Get types failed, due to exception ");
                        e.printStackTrace();
                    }

                    // Updates bizTypes cache
                    try {
                        String[] bizTypesInHd = FieldCache.DEFAULT.getStrings(LuceneSearchCommandRefactor
                                .getHdIndexSearcher().getIndexReader(), StringConst.P_TYPE);
                        LuceneSearchCommandRefactor.setBizTypesInHd(bizTypesInHd);
                    } catch (IOException e) {
                        log.error("Get bizTypes failed, due to exception ");
                        e.printStackTrace();
                    }

                    // Updates tags cache
                    try {
                        String[] tagsInHd = FieldCache.DEFAULT.getStrings(LuceneSearchCommandRefactor
                                .getHdIndexSearcher().getIndexReader(), StringConst.TAGS);
                        LuceneSearchCommandRefactor.setTagsInHd(tagsInHd);
                    } catch (IOException e) {
                        log.error("Get bizTypes failed, due to exception ");
                        e.printStackTrace();
                    }

                    // Update MaxHitCountByTags map
                    LuceneSearchCommandRefactor.setMaxHitCountByTagsMap(infoCommand
                            .getMaxHitCountByTags(StringConst.DOCUMENTARY_CHINESE));

                }
            } catch (Exception e) {
                log.error("refresh index cache failed @ " + new Date());
                e.printStackTrace();
            }
        }
    }

    public void resetLastIndexTime() {
        setLastIndexDateTime(StringConst.HD_LAST_INDEX_TIME_HEADERS, null);
        setLastIndexDateTime(StringConst.HD_LAST_INDEX_TIME_INFOS, null);
        setLastIndexDateTime(StringConst.HD_LAST_INDEX_TIME_PERSONS, null);
        setLastIndexDateTime(StringConst.LAST_INDEX_TIME_HEADERS, null);
        setLastIndexDateTime(StringConst.LAST_INDEX_TIME_INFOS, null);
        setLastIndexDateTime(StringConst.LAST_INDEX_TIME_KEYS, null);
    }

    private void doIndexForHeaders(boolean isHd) {
        Date maxUpdateTimeForIndex = null;
        // Update data to index for headers
        log.info("Indexing headers starting..... @ " + new Date());

        Date lastIndexTime = null;
        if (isHd) {
            lastIndexTime = getLastIndexDateTime(StringConst.HD_LAST_INDEX_TIME_HEADERS);
        } else {
            lastIndexTime = getLastIndexDateTime(StringConst.LAST_INDEX_TIME_HEADERS);
        }

        List<InfoHeader> headers = null;
        if (lastIndexTime != null) {
            headers = infoHeaderCommand.getHeadersByUpdateTime(lastIndexTime);
        } else {
            headers = infoHeaderCommand.getAllValidHeaders();
        }

        if (headers != null && headers.size() > 0) {
            maxUpdateTimeForIndex = headers.get(headers.size() - 1).getUpdate_time_for_index();
            // Convert to lucene object
            if (isHd) {
                List<HdInfoLucene> hdInfos = infoHeaderCommand.convertToHdInfoLucene(headers);
                doIndex(hdInfos);

                // Save last index time
                setLastIndexDateTime(StringConst.HD_LAST_INDEX_TIME_HEADERS, maxUpdateTimeForIndex);
                shouldReopenHdInfo = true;
            } else {
                List<Info> infos = infoHeaderCommand.convertToInfoLucene(headers);
                doIndex(infos);

                // Save last index time
                setLastIndexDateTime(StringConst.LAST_INDEX_TIME_HEADERS, maxUpdateTimeForIndex);
                shouldReopenInfo = true;
            }
        }
        log.info("Indexing headers finished ..... @ " + new Date());
    }

    private void doIndexForInfos(boolean isHd, Set<String> relatedPersonIds) {
        // In order to avoid outOfMemory, just write 10000/5000 info to index
        // for each time.
        int indexLimit = 10000;

        Date lastIndexTime = null;
        if (isHd) {
            lastIndexTime = getLastIndexDateTime(StringConst.HD_LAST_INDEX_TIME_INFOS);
            indexLimit = 5000;
        } else {
            lastIndexTime = getLastIndexDateTime(StringConst.LAST_INDEX_TIME_INFOS);
        }
        // Update data to index for info
        if (lastIndexTime != null) {
            // Increment indexing Gets increased info
            log.info(" Increment indexing info starting..... @ " + new Date());
            List<Info> infos = infoCommand.getInfosByUpdateTimeForIndex(lastIndexTime);
            if (infos != null && infos.size() > 0) {
                Date maxIndexTime = infos.get(infos.size() - 1).getUpdateTimeForIndex();

                List<Info> indexInfos = new ArrayList<Info>();
                for (int i = 0; i < infos.size(); i++) {
                    indexInfos.add(infos.get(i));
                    if (indexInfos.size() == indexLimit || i == infos.size() - 1) {
                        if (isHd) {
                            shouldReopenHdInfo = true;
                            // Convert to HdInfoLucene
                            List<HdInfoLucene> list = infoCommand.convertToHdInfoLucene(indexInfos);
                            // Start index
                            doIndex(list);
                            // gets person ids
                            if (relatedPersonIds != null) {
                                relatedPersonIds.addAll(personCommand.getPersonIdsByInfoIds(indexInfos));
                            }
                            setLastIndexDateTime(StringConst.HD_LAST_INDEX_TIME_INFOS, maxIndexTime);
                        } else {
                            shouldReopenInfo = true;
                            List<Info> list = infoCommand.convertToLuceneInfo(indexInfos);
                            doIndex(list);
                            setLastIndexDateTime(StringConst.LAST_INDEX_TIME_INFOS, maxIndexTime);
                        }

                        // Sleep 5s for waiting last index finished.
                        if (indexInfos.size() == indexLimit) {
                            try {
                                Thread.sleep(5000);
                            } catch (InterruptedException e) {
                                e.printStackTrace();
                            }
                        }

                        // reset indexInfos list.
                        indexInfos = new ArrayList<Info>();
                    }
                }
            }
            log.info("Increment indexing info finished ..... @ " + new Date());
        } else {
            log.info("Full indexing info starting..... @ " + new Date());
            // full indexing for info
            int count = infoCommand.getOnLineInfosCount();
            int start = 0;
            do {
                List<Info> infos = infoCommand.getOnLineInfos(start, indexLimit);
                if (infos != null && infos.size() > 0) {
                    Date maxIndexTime = infos.get(infos.size() - 1).getUpdateTimeForIndex();
                    if (isHd) {
                        shouldReopenHdInfo = true;
                        // Convert to HdInfoLucene
                        List<HdInfoLucene> list = infoCommand.convertToHdInfoLucene(infos);
                        // Start index
                        doIndex(list);

                        setLastIndexDateTime(StringConst.HD_LAST_INDEX_TIME_INFOS, maxIndexTime);
                    } else {
                        shouldReopenInfo = true;
                        List<Info> list = infoCommand.convertToLuceneInfo(infos);
                        doIndex(list);

                        setLastIndexDateTime(StringConst.LAST_INDEX_TIME_INFOS, maxIndexTime);
                    }
                }
                start += indexLimit;
            } while (start < count);
            log.info("Full indexing info finished ..... @ " + new Date());
        }
    }

    private void doIndexForKeySubject() {
        // In order to avoid outOfMemory, just write 10000 info to index for
        // each
        // time.
        final int indexLimit = 10000;

        Date lastIndexTime = getLastIndexDateTime(StringConst.LAST_INDEX_TIME_KEYS);
        // Update data to index for key subject
        if (lastIndexTime != null) {
            // Gets increased keySubjects
            log.info("Increment indexing key subject starting..... @ " + new Date());
            List<KeySubject> keys = keySubjectCommand.getKeySubjectsByUpdateTimeForIndex(lastIndexTime);
            if (keys != null && keys.size() > 0) {
                shouldReopenKey = true;
                doIndex(keys);
                Date maxIndexTime = keys.get(keys.size() - 1).getUpdateTimeForIndex();
                setLastIndexDateTime(StringConst.LAST_INDEX_TIME_KEYS, maxIndexTime);
            }
            log.info("Increment indexing key subject finished ..... @ " + new Date());
        } else {
            // full indexing for keySubject
            log.info("Full indexing key subject starting..... @ " + new Date());
            int count = keySubjectCommand.getOnLineKeySubjectsCount();
            int start = 0;
            do {
                List<KeySubject> keys = keySubjectCommand.getOnLineKeySubjects(start, indexLimit);
                if (keys != null && keys.size() > 0) {
                    shouldReopenKey = true;
                    doIndex(keys);
                    Date maxIndexTime = keys.get(keys.size() - 1).getUpdateTimeForIndex();
                    setLastIndexDateTime(StringConst.LAST_INDEX_TIME_KEYS, maxIndexTime);
                }
                start += indexLimit;
            } while (start < count);

            log.info("Full indexing key subject finished ..... @ " + new Date());
        }
    }

    private void doIndexForPerson() {
        // In order to avoid outOfMemory, just write 200 info to index for each
        // time.
        final int indexLimit = 2000;

        Date lastIndexTime = getLastIndexDateTime(StringConst.HD_LAST_INDEX_TIME_PERSONS);
        // Update data to index for persons
        if (lastIndexTime == null) {
            // Set hdLastIndexTimeOfPerson to 2013-01-01, due to this change
            // is more later than 2013-01-01
            Calendar c = Calendar.getInstance();
            c.set(2013, 0, 1);
            lastIndexTime = c.getTime();
        }

        int count = personCommand.getPersonCountByUpdateTime(lastIndexTime);
        int start = 0;
        do {

            List<HdPersonLucene> persons = personCommand.getPersonByUpdateTime(lastIndexTime, start, indexLimit);
            if (persons != null && persons.size() > 0) {
                shouldReopenHdInfo = true;

                // do index
                doIndex(persons);

                // gets max update time
                Date maxIndexTime = persons.get(persons.size() - 1).getUpdateTime();
                setLastIndexDateTime(StringConst.HD_LAST_INDEX_TIME_PERSONS, maxIndexTime);
            }
            start += indexLimit;
        } while (start < count);

        log.info("Person indexing finished ..... @ " + new Date());

    }

    private <T> void doIndex(final List<T> list) {
        if (list != null && list.size() != 0) {
            getCompassTemplate().execute(new CompassCallback<String>() {
                public String doInCompass(CompassSession compassSession) throws CompassException {
                    Set<String> offlineCodes = new HashSet<String>();
                    try {
                        int count = 0;
                        for (T obj : list) {
                            if (obj instanceof Info) {
                                Info info = (Info) obj;
                                log.debug("Info " + info.getCode() + " is being indexed and its update flag is "
                                        + info.getUpdateFlag());
                                if (info.getUpdateFlag() == 0 || info.getUpdateFlag() == 1) {
                                    compassSession.save(info);
                                } else if (info.getUpdateFlag() == 2) {
                                    compassSession.delete(info);
                                    offlineCodes.add(info.getCode());
                                }
                            } else if (obj instanceof KeySubject) {
                                KeySubject sb = (KeySubject) obj;
                                if (sb.getUpdateFlag() == 0) {
                                    if (compassSession.getResource(KeySubject.class, sb.getId()) == null) {
                                        compassSession.create(sb);
                                    }
                                } else if (sb.getUpdateFlag() == 1) {
                                    compassSession.delete(KeySubject.class, sb.getId());
                                    compassSession.save(sb);
                                }
                            } else if (obj instanceof HdPersonLucene) {
                                HdPersonLucene person = (HdPersonLucene) obj;

                                if (person.getUpdate_flag() == 1) {
                                    compassSession.delete(person);
                                } else if (person.getUpdate_flag() == 0) {
                                    compassSession.save(person);
                                }
                            } else if (obj instanceof HdInfoLucene) {
                                HdInfoLucene hdInfo = (HdInfoLucene) obj;
                                if (hdInfo.getUpdateFlag() == 0 || hdInfo.getUpdateFlag() == 1) {
                                    compassSession.save(hdInfo);
                                } else if (hdInfo.getUpdateFlag() == 2) {
                                    compassSession.delete(hdInfo);
                                    offlineCodes.add(hdInfo.getCode());
                                }
                            } else {
                                log.error("Indexed unknow type " + obj.getClass().getName());
                            }

                            // Fix bug,
                            // Exception: "Failed to add job xxx after [10000ms] and backlog size [100]".
                            // flush this session when count>=100 for each time
                            count++;
                            if (count >= 100) {
                                compassSession.flush();
                                count = 0;
                            }
                        }
                    } catch (Exception e) {
                        log.error(" update data to index failed @ " + new Date());
                        e.printStackTrace();
                    }

                    if (offlineCodes != null && offlineCodes.size() > 0) {
                        // Remove offlines info from cache
                        infoCommand.removeOfflineCodesFromTopInfoCache(offlineCodes);
                    }
                    return null;
                }
            });
        }
    }

    private Date getLastIndexDateTime(String key) {
        PropertiesConfiguration config;
        try {
            config = new PropertiesConfiguration(lastIndexTimeFile);
        } catch (ConfigurationException e) {
            log.error("Load last index time file failed");
            log.error(e.toString());
            e.printStackTrace();
            return null;
        }
        config.setReloadingStrategy(new FileChangedReloadingStrategy());
        String value = config.getString(key);
        if (value != null && !StringConst.EMPTY.equals(value.trim())) {
            try {
                return DateUtil.convertStringToDate(LAST_INDEX_TIME_FORMAT, value);
            } catch (Exception e) {
                log.error("Get last index time failed from lastIndexTimeFile @ " + new Date());
                e.printStackTrace();
            }
        }

        return null;
    }

    private void setLastIndexDateTime(String key, Date date) {
        String lastIndexTime = StringConst.EMPTY;
        if (date != null) {
            lastIndexTime = DateUtil.convertDateToString(LAST_INDEX_TIME_FORMAT, date);
        }

        PropertiesConfiguration config;
        try {
            config = new PropertiesConfiguration(lastIndexTimeFile);
            if (config.containsKey(key)) {
                config.setProperty(key, lastIndexTime);
            } else {
                config.addProperty(key, lastIndexTime);
            }
            config.save();
        } catch (ConfigurationException e) {
            log.error("Load last index time file failed");
            log.error(e.toString());
            e.printStackTrace();
        }
    }

    // Start getter/setter methods

    public String getSortType() {
        return sortType;
    }

    public void setSortType(String sortType) {
        this.sortType = sortType;
    }

    public String getLastIndexTimeFile() {
        return lastIndexTimeFile;
    }

    public void setLastIndexTimeFile(String lastIndexTimeFile) {
        this.lastIndexTimeFile = lastIndexTimeFile;
    }

    public Compass getCompass() {
        return compass;
    }

    public void setCompass(Compass compass) {
        this.compass = compass;
    }

    public CompassTemplate getCompassTemplate() {
        return compassTemplate;
    }

    public void setCompassTemplate(CompassTemplate compassTemplate) {
        this.compassTemplate = compassTemplate;
    }

    public boolean isCanStartIncrementIndexThread() {
        return canStartIncrementIndexThread;
    }

    public void setCanStartIncrementIndexThread(boolean canStartIncrementIndexThread) {
        this.canStartIncrementIndexThread = canStartIncrementIndexThread;
    }

    public InfoCommand getInfoCommand() {
        return infoCommand;
    }

    public void setInfoCommand(InfoCommand infoCommand) {
        this.infoCommand = infoCommand;
    }

    public KeySubjectCommand getKeySubjectCommand() {
        return keySubjectCommand;
    }

    public void setKeySubjectCommand(KeySubjectCommand keySubjectCommand) {
        this.keySubjectCommand = keySubjectCommand;
    }

    public PersonCommand getPersonCommand() {
        return personCommand;
    }

    public void setPersonCommand(PersonCommand personCommand) {
        this.personCommand = personCommand;
    }

    public InfoHeaderCommand getInfoHeaderCommand() {
        return infoHeaderCommand;
    }

    public void setInfoHeaderCommand(InfoHeaderCommand infoHeaderCommand) {
        this.infoHeaderCommand = infoHeaderCommand;
    }

    public BizTypeManager getBizTypeManager() {
        return bizTypeManager;
    }

    public void setBizTypeManager(BizTypeManager bizTypeManager) {
        this.bizTypeManager = bizTypeManager;
    }

    public SystemCache getSystemCache() {
        return systemCache;
    }

    public void setSystemCache(SystemCache systemCache) {
        this.systemCache = systemCache;
    }

    // End getter/setter methods
}
